package ar.edu.itba.pod.legajo48062;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import ar.edu.itba.balance.api.AgentsBalancer;
import ar.edu.itba.balance.api.AgentsTransfer;
import ar.edu.itba.balance.api.NodeAgent;
import ar.edu.itba.balance.api.NotCoordinatorException;
import ar.edu.itba.node.Node;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.node.api.ClusterAdministration;
import ar.edu.itba.pod.thread.CleanableThread;

public class DistributedAgentsBalancer extends UnicastRemoteObject implements AgentsBalancer {
	private static final long serialVersionUID = 1L;

	private DistributedSimulation node;
	private Cluster cluster;
	private NodeInformation coordinator;
	private AtomicBoolean myElectionRunning;
	private Semaphore lockElection;
	private CountDownLatch lockCoordinator;
	
	private Map<NodeInformation,Long> electionMessages;
	private Map<NodeInformation,Long> coordinatorMessages;
	
	protected DistributedAgentsBalancer(ClusterAdministration cluster) throws RemoteException {
		super();
		this.cluster=(Cluster) cluster;
		myElectionRunning=new AtomicBoolean(false); // PARA SABER SI ESTOY CORRIENDO UNA ELECCION
		lockElection=new Semaphore(0); // PARA BLOQUEARME HASTA QUE TERMINE LA ELECCION ACTIVA
		lockCoordinator=new CountDownLatch(1); // PARA BLOQUEARME HASTA QUE SE DECIDA EL COORDINADOR
		electionMessages=new ConcurrentHashMap<NodeInformation, Long>();
		coordinatorMessages=new ConcurrentHashMap<NodeInformation, Long>();
		
		new CleanerThread("cleanElectionMessagesThread", electionMessages, Host.electionMessagesRemoveFrequency).start();
		new CleanerThread("cleanCoordinatorMessagesThread", coordinatorMessages, Host.coordinatorMessagesRemoveFrequency).start();
	}
	
	public void setNode(DistributedSimulation node){
		this.node=(DistributedSimulation) node;
	}
	
	// DEVUELVE EN EL PARAMETRO AGENTSBYNODE UNA LISTA DE TODOS LOS NODOS Y LA CANTIDAD DE AGENTES DE CADA UNO ORDENADO ASCENDENTEMENTE POR LA CANTIDAD DE AGENTES. 
	// DEVUELVE EN LOS PARAMETROS AGENTSCOUNT Y NODESCOUNT LA CANTIDAD TOTAL DE AGENTES Y DE NODOS
	// EXCLUDENODE SIRVE PARA EL SHUTDOWN YA QUE NECESITO LAS ESTADISTICAS SIN EL NODO QUE ACABA DE BAJARSE
	private void getStatistics(final List<NodeAgentsAmount> agentsByNode,final AtomicInteger agentsCount,final AtomicInteger nodesCount, NodeInformation exceptNode){
		agentsByNode.clear();
		if (exceptNode!=null && exceptNode.equals(cluster.me())){
			agentsCount.set(0);
			nodesCount.set(0);
		}else{
			agentsByNode.add(new NodeAgentsAmount(cluster.me(), node.agentsRunning()));
			agentsCount.set(node.agentsRunning());
			nodesCount.set(1);
		}
		
		new Communication<AgentsTransfer>(cluster,Node.AGENTS_TRANSFER){
			@Override
			public void remoteCall(AgentsTransfer stub, NodeInformation node) throws RemoteException {
				int count=stub.getNumberOfAgents();
				agentsByNode.add(new NodeAgentsAmount(node, count));
				agentsCount.addAndGet(count);
				nodesCount.incrementAndGet();
			}
		}.broadcast(exceptNode, true, true);
		
		Collections.sort(agentsByNode);
	}
	
	private void printAgentsByNode(List<NodeAgentsAmount> agentsByNode){
		for (NodeAgentsAmount n : agentsByNode){
			System.out.println("NODO: "+n.getNode().id()+" - "+n.getAmount()+" AGENTES");
		}
	}
	
	private boolean isBalanced(List<NodeAgentsAmount> agentsByNode){
		return (agentsByNode.get(agentsByNode.size()-1).getAmount() - agentsByNode.get(0).getAmount() <= 1); 
	}
	
	// PRECONDICIONES: EL ALGORITMO ASUME QUE TODOS LOS NODOS ESTAN BALANCEADOS, A EXCEPCION DE UNO QUE TIENE MENOS AGENTES QUE EL RESTO
	// ESTE METODO SE LLAMA AL AGREGAR UN NUEVO NODO, POR LO QUE ESTO SE CUMPLE Y EN PARTICULAR EL NUEVO NODO TIENE 0 AGENTES
	// ALGORITMO:
	// 1) CALCULA CUANTOS AGENTES DEBERA TENER FINALMENTE CADA NODO (COMO PUEDE NO SER ENTERO LA VARIABLE REST INDICA CUANTOS NODOS TIENEN +1 AGENTE QUE EL RESTO)
	// 2) SELECCIONA EL NODO CON MENOS AGENTES (EL NUEVO, QUE RECIBIRA LOS AGENTES DEL RESTO). ESTE NODO ES EL PRIMERO DE LA LISTA AGENTSBYNODE
	// 3) RECORRE TODOS LOS NODOS Y LES SACA LOS AGENTES QUE CORRESPONDAN PARA QUE QUEDEN CON EL NUMERO ADECUADO CALCULADO ANTERIORMENTE
	// 4) CORRE TODOS LOS AGENTES RECOLECTADOS EN EL NODO CON MENOS AGENTES
	@SuppressWarnings("unchecked")
	public void newNodeBalance(){
		synchronized (DistributedAgentsBalancer.class) {
			NodeInformation nodeWithLessAgents;
			final List<NodeAgentsAmount> agentsByNode = new ArrayList<NodeAgentsAmount>();
			final AtomicInteger agentsCount= new AtomicInteger();
			final AtomicInteger nodesCount= new AtomicInteger();
			double average;
			int averageInt;
			boolean isBalanced;
			int rest;
			final List<NodeAgent> agentsToTransfer=new ArrayList<NodeAgent>();
	
			System.out.println("ESTADISTICAS PRE-NEWNODEBALANCE");
			getStatistics(agentsByNode, agentsCount, nodesCount, null);
			nodeWithLessAgents = agentsByNode.get(0).getNode();
			average = (double) agentsCount.get() / (double) nodesCount.get();
			averageInt = agentsCount.get() / nodesCount.get();
			rest = agentsCount.get() % nodesCount.get();
			isBalanced=isBalanced(agentsByNode);
			
			printAgentsByNode(agentsByNode);
			System.out.println("MIN-NODE "+nodeWithLessAgents.id()+" - AVERAGE: "+average+" - REST: "+rest+" - ISBALANCED: "+isBalanced);
			
			if (!isBalanced){
				for (NodeAgentsAmount agentAmount : agentsByNode){
					int agents = agentAmount.getAmount();
					NodeInformation n = agentAmount.getNode();
					
					if (agents>averageInt){
						final int agentsToGet;
						
						if (rest>0){
							agentsToGet=agents-averageInt-1;
							rest--;
						}else{
							agentsToGet=agents-averageInt;
						}
						
						if (agentsToGet>0){
							List<NodeAgent> nodeAgents = (List<NodeAgent>) new Communication<AgentsTransfer>(cluster,Node.AGENTS_TRANSFER){
								@Override
								public Object remoteCallWithReturn(AgentsTransfer stub, NodeInformation node) throws Exception {
									return stub.stopAndGet(agentsToGet);
								}
							}.unicastWithReturn(n,true);
							
							if (nodeAgents==null){
								System.out.println("HUBO UN ERROR GRAVE EL HACER UN STOP AND GET DE "+agentsToGet+" AGENTES DEL NODO "+n.id());
							}else{
								System.out.println("LE SAQUE "+agentsToGet+" AGENTES AL NODO "+n.id());
								agentsToTransfer.addAll(nodeAgents);
								if (agentsToTransfer.size()==averageInt){
									System.out.println("TERMINE DE JUNTAR LOS AGENTES PARA EL NUEVO NODO");
									break;
								}
							}
						}
					}
				}
				boolean success = new Communication<AgentsTransfer>(cluster,Node.AGENTS_TRANSFER){
					@Override
					public void remoteCall(AgentsTransfer stub, NodeInformation node) throws RemoteException {
						stub.runAgentsOnNode(agentsToTransfer);
					}
				}.unicast(nodeWithLessAgents, true);	
				
				if (!success){
					System.out.println("HUBO UN ERROR GRAVE AL HACERLE RUNAGENTS AL NODO "+nodeWithLessAgents.id());
				}else{
					System.out.println("SE LE TRANSFIRIERON "+agentsToTransfer.size()+" AGENTES AL NODO "+nodeWithLessAgents.id());
				}
			
				System.out.println("ESTADISTICAS POST-NEWNODEBALANCE");
				getStatistics(agentsByNode, agentsCount, nodesCount, null);
				isBalanced=isBalanced(agentsByNode);
				printAgentsByNode(agentsByNode);
				System.out.println("ISBALANCED: "+isBalanced);
			}
		}
	}
	
	// CON LA LISTA DE AGENTES POR NODO ORDENADA OBTIENE EL NODO CON MENOS AGENTES (EL PRIMERO) Y A ESE LE CORRE EL AGENTE RECIBIDO
	@Override
	public void addAgentToCluster(NodeAgent agent) throws RemoteException, NotCoordinatorException {
		if (!getCoordinator().equals(cluster.me())){ throw new NotCoordinatorException(coordinator); }
		
		synchronized (DistributedAgentsBalancer.class) {
			List<NodeAgentsAmount> agentsByNode = new ArrayList<NodeAgentsAmount>();
			AtomicInteger agentsCount= new AtomicInteger();
			AtomicInteger nodesCount= new AtomicInteger();

			getStatistics(agentsByNode, agentsCount, nodesCount, null);
			NodeInformation nodeWithLessAgents = agentsByNode.get(0).getNode();
			System.out.println("EL NODO CON MENOS AGENTES ES:"+nodeWithLessAgents.id());
			final List<NodeAgent> agents=new ArrayList<NodeAgent>();
			agents.add(agent);
			boolean success = new Communication<AgentsTransfer>(cluster,Node.AGENTS_TRANSFER){
				@Override
				public void remoteCall(AgentsTransfer stub, NodeInformation node) throws RemoteException {
					stub.runAgentsOnNode(agents);
				}
			}.unicast(nodeWithLessAgents, true);
			if (success){
				System.out.println("YA LE CORRI EL AGENTE A "+nodeWithLessAgents.id());
			}else{
				System.out.println("ERROR GRAVE: NO LE PUDE CORRER EL AGENTE A "+nodeWithLessAgents.id()+". TAL VEZ QUEDO AGENTE PERDIDO");
			}
		}
	}

	// METODO PARA HACER DE MANERA ATOMICA EL CHEQUEO DE MENSAJES DE ELECTION Y COORDINATOR Y EL AGREGADO DEL MENSAJE A LA LISTA
	public boolean checkAndPutInMap(Map<NodeInformation,Long> map, NodeInformation node, Long timestamp){
		synchronized (Communication.class) {
			if (!(map.containsKey(node) && map.get(node).equals(timestamp))){
				map.put(node,timestamp);
				return false;
			}else{
				return true;
			}
		}
	}
	
	@Override
	public void bullyCoordinator(NodeInformation node, long timestamp) throws RemoteException {
		final NodeInformation sourceNode = node;
		final long sourceTimestamp = timestamp;
		
		if (!checkAndPutInMap(coordinatorMessages, node, timestamp)){
			coordinator=node;
			lockCoordinator.countDown();
			System.out.println("Proclamo coordinador a "+coordinator.id());

			if (cluster.me().equals(coordinator)){
				new CleanableThread("NewNodeBalance") {
					@Override
					public void run() {
						newNodeBalance();
					}
				}.start();
			}
			new CleanableThread("Election") {
				public void run() {
					new Communication<AgentsBalancer>(cluster,Node.AGENTS_BALANCER){
						@Override
						public void remoteCall(AgentsBalancer stub, NodeInformation node) throws RemoteException {
							stub.bullyCoordinator(sourceNode, sourceTimestamp);
						}
					}.broadcast(sourceNode,true, true);
				};
			}.start();
		}
	}

	@Override
	public void bullyElection(final NodeInformation node, final long timestamp)	throws RemoteException {
		final boolean higherId=(cluster.me().id().compareTo(node.id())>0);
		final boolean myElection=node.equals(cluster.me());
		final boolean lowerId=(!higherId && !myElection);

		if (!checkAndPutInMap(electionMessages, node, timestamp)){
			coordinator=null;
			lockCoordinator=new CountDownLatch(1);
			if (higherId){
				new Communication<AgentsBalancer>(cluster,Node.AGENTS_BALANCER){
					@Override
					public void remoteCall(AgentsBalancer stub, NodeInformation node) throws RemoteException {
						stub.bullyOk(cluster.me());
					}
				}.unicast(node, true);
			}
			
			new CleanableThread("Election") {
				@Override
				public void run() {
					if (lowerId){
						new Communication<AgentsBalancer>(cluster,Node.AGENTS_BALANCER){
							@Override
							public void remoteCall(AgentsBalancer stub, NodeInformation nodeIterator) throws RemoteException {
								stub.bullyElection(node, timestamp);
							}
						}.broadcast(node,true, true);
					}else{
						if (myElectionRunning.compareAndSet(false, true)){
							lockElection=new Semaphore(0);
							final long currentTime = Host.getTime();
							new Communication<AgentsBalancer>(cluster,Node.AGENTS_BALANCER){
								@Override
								public void remoteCall(AgentsBalancer stub, NodeInformation node) throws RemoteException {
									stub.bullyElection(cluster.me(), currentTime);
								}
							}.broadcast(cluster.me(),true, true);
							
							try {
								boolean anyOk = lockElection.tryAcquire(5, TimeUnit.SECONDS);
								if (!anyOk){
									bullyCoordinator(cluster.me(), Host.getTime());
								}
							} catch (Exception e) {
								e.printStackTrace();
							}
							myElectionRunning.set(false);
						}						
					}
				}
			}.start();
		}
	}

	@Override
	public void bullyOk(NodeInformation node) throws RemoteException {
		lockElection.release();
	}

	// ALGORITMO:
	// 1) CONSIGO LA LISTA DE AGENTES POR NODO ORDENADA (SIN CONTEMPLAR AL NODO QUE HIZO SHUTDOWN)
	// 2) CALCULO EL PROMEDIO DE AGENTES A ENVIAR A CADA NODO (ESTO PUEDE NO SER ENTERO SE USA LA VARIABLE REST)
	// 3) PARA CADA NODO ASIGNO LOS AGENTES CORRESPONDIENTES SACANDO DE LA LISTA DE AGENTES ENVIADOS
	@Override
	public void shutdown(final List<NodeAgent> agents) throws RemoteException, NotCoordinatorException {
		if (!getCoordinator().equals(cluster.me())){ throw new NotCoordinatorException(coordinator); }

		synchronized (DistributedAgentsBalancer.class) {
			NodeInformation disconnectNode = agents.get(0).node();
			int agentsToTransfer = agents.size();
			final List<NodeAgentsAmount> agentsByNode = new ArrayList<NodeAgentsAmount>();
			final AtomicInteger agentsCount= new AtomicInteger();
			final AtomicInteger nodesCount= new AtomicInteger();
			final int averageInt;
			int rest;

			getStatistics(agentsByNode, agentsCount, nodesCount, disconnectNode);
			
			System.out.println("ME LLEGO UN SHUTDOWN DE "+agents.size()+" AGENTES DEL NODO "+disconnectNode.id());
			System.out.println("ESTADISTICAS PRE-SHUTDOWNBALANCE");
			printAgentsByNode(agentsByNode);
			
			averageInt = agentsToTransfer / nodesCount.get();
			rest = agentsToTransfer % nodesCount.get();
			
			for (NodeAgentsAmount node : agentsByNode){
				final int amountToTransfer;
				if (rest==0){
					amountToTransfer=averageInt;
				}else{
					amountToTransfer=averageInt+1;
					rest--;
				}
				if (amountToTransfer>0){
					final List<NodeAgent> agentsAux = new ArrayList<NodeAgent>();
					for (int i=0;i<amountToTransfer;i++){
						agentsAux.add(agents.get(0));
						agents.remove(0);
					}

					boolean success = new Communication<AgentsTransfer>(cluster,Node.AGENTS_TRANSFER){
						@Override
						public void remoteCall(AgentsTransfer stub, NodeInformation node) throws RemoteException {
							stub.runAgentsOnNode(agentsAux);
						}
					}.unicast(node.getNode(),true);
					if (success){
						System.out.println("YA LE CORRI EL AGENTE A "+node.getNode().id());
					}else{
						System.out.println("ERROR GRAVE: NO LE PUDE CORRER LOS AGENTE A "+node.getNode().id());
					}
				}
			}
			
			getStatistics(agentsByNode, agentsCount, nodesCount, disconnectNode);
			System.out.println("ESTADISTICAS POST-SHUTDOWNBALANCE");
			printAgentsByNode(agentsByNode);
		}
	}
	
	// IMPORTANTE: ESTE METODO ES BLOQUEANTE, HASTA QUE NO HAY UN COORDINADOR DEFINIDO NO RETORNA!
	// SI SE ESTA CORRIENDO UNA ELECCION SE ESPERA A QUE TERMINE, SI NO SE ESTA CORRIENDO UNA ELECCION SE EMPIEZA UNA
	public NodeInformation getCoordinator(){
		while(coordinator==null){
			try{
				System.out.println("No tenia coordinador asi que lanzo una eleccion");
				bullyElection(cluster.me(), Host.getTime());
				boolean result = lockCoordinator.await(5,TimeUnit.SECONDS);
				if (result){
					System.out.println("Termino la eleccion");
				}else{
					System.out.println("Se arranco una eleccion pero nunca llego un bullyCoordinator");
				}
			}catch (Exception e) {
				e.printStackTrace();
			}
		}
		return coordinator;
	}
}
